package com.heima.media.listener;

import com.alibaba.fastjson.JSON;
import com.heima.media.entity.WmNews;
import com.heima.media.service.IAuditService;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Service
public class PublishListener {

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private IAuditService auditService;

    @PostConstruct
    public void listen() throws InterruptedException {

        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {

                // 获取阻塞队列
                RBlockingQueue<String> queue = redissonClient.getBlockingQueue("publish-article");
                // 获取延迟队列
                RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(queue);
                // 初始化通道
                delayedQueue.offer("init", 2, TimeUnit.SECONDS);
                String poll = null;

                try {
                    while (true) {

                        poll = queue.poll(1, TimeUnit.MINUTES);

                        if (poll != null && !poll.equals("init")) {

                            System.out.println("接收到消息" + poll + new Date());

                            WmNews wmNews = JSON.parseObject(poll, WmNews.class);

                            auditService.audit(wmNews);
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        thread.start();
    }
}
